package f7richfunclass_lifecycle;

import flinkemp.Emp;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class RichfunLife {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(1);
        DataStreamSource<Emp> empstream = env.fromElements(
                new Emp(1000,1001,"zs","Analyst",6900.0,10),
                new Emp(1000,1001,"zs","Analyst",6900.0,10),
                new Emp(1000,1001,"zs","Analyst",6900.0,10),
                new Emp(1000,1001,"zs","Analyst",6900.0,10),
                new Emp(1000,1001,"zs","Analyst",6900.0,10),
                new Emp(1000,1001,"zs","Analyst",6900.0,10),
                new Emp(1000,1001,"zs","Analyst",6900.0,10)
        );
        empstream.print();
        empstream.map(new RichMapFunction<Emp, String>() {
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                System.out.printf("索引为 " + getRuntimeContext().getIndexOfThisSubtask() + " 的任务开始");//获取当前任务在并行环境中的索引
            }
            @Override
            public String map(Emp emp) throws Exception {
                return emp.toString();
            }
            @Override
            public void close() throws Exception {
                super.close();
                System.out.printf("索引为 " + getRuntimeContext().getIndexOfThisSubtask() + " 的任务结束，资源释放");// 获取当前任务在并行环境中的索引
            }
        }).print();

        env.execute();
    }
}
